package com.wei.weblog.mr;

import java.io.IOException;
import java.net.URI;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Locale;
import java.util.UUID;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Reducer;

import com.wei.weblog.mr.WebLogVisitJob.WebLogVisitMapper;
import com.wei.weblog.mr.WebLogVisitJob.WebLogVisitReduce;
import com.wei.weblog.pre.WebLogParser;

/**
 * 单session访问总结
 * sessionid  start-time   out-time   start-page   out-page   pagecounts  ......
 * @author Administrator
 *
 */
public class WebLogVisitSummaryJob {
	
	static class VisitSummaryMapper extends Mapper<LongWritable, Text, Text, WebLogInfo>{

		WebLogInfo info=new WebLogInfo();
		Text text=new Text();
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			String[] values = line.split("\001");
			/**
			 * 160false3340e59f-e4a8-4d11-899b-cac8fc3bf9b9180.153.163.190-2013-09-18 09:37:41"GET
			 * /finance-rhive-repurchase/
			 * HTTP/1.1"20045271"-""Mozilla/4.0"
			 */
			if (values.length>12) {
				String session = values[3];
				String remoteIp = values[4];
				String remoteUserInfo = values[5];
				String requestTime = values[6];
				String requestMethod = values[7];
				String requestUrl = values[8];
				String requestProtocol = values[9];
				String responseCode = values[10];
				String bodySendBytes = values[11];
				String refererUrl = values[12];
				String userAgent = values[13];
				text.set(session);
				info.set(true, remoteIp, remoteUserInfo, requestTime, requestUrl, requestMethod, requestProtocol, responseCode, bodySendBytes, refererUrl, userAgent);
				context.write(text, info);
			}
		}
		
	}
	
	static class VisitSummaryReduce extends Reducer<Text, WebLogInfo, Text, Text>{

		Text key=new Text();
		Text value=new Text();
		@Override
		protected void reduce(Text text, Iterable<WebLogInfo> iterable,Context context)
				throws IOException, InterruptedException {
			List<WebLogInfo> webLogInfoList = new ArrayList<>();
			for (WebLogInfo webLogInfo : iterable) {
				WebLogInfo info = new WebLogInfo();
				try {
					BeanUtils.copyProperties(info, webLogInfo);
					webLogInfoList.add(info);
				} catch (Exception e) {
					continue;
				}
			}
			Collections.sort(webLogInfoList, new Comparator<WebLogInfo>() {
				@Override
				public int compare(WebLogInfo o1, WebLogInfo o2) {
					String requestTime2 = o1.getRequestTime();
					String requestTime3 = o2.getRequestTime();
					SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss", Locale.US);
					try {
						Date parse = df.parse(requestTime2);
						Date parse2 = df.parse(requestTime3);
						return parse.compareTo(parse2);
					} catch (ParseException e) {
						return 0;
					}
				}
			});
			String stayTime="60";
			String outPage="";
			String outTime="";
			int size = webLogInfoList.size();
			int pageVisits=size;
			WebLogInfo webLogInfo0 = webLogInfoList.get(0);
			if (size==1) {
				StringBuffer buffer=new StringBuffer();
				outPage=webLogInfo0.getRequestUrl();
				outTime=webLogInfo0.getRequestTime();
				buffer.append(pageVisits+"\001"+stayTime+"\001"+outPage+"\001"+outTime+"\001");
				buffer.append(webLogInfo0.toString());
				value.set(buffer.toString());
				context.write(key, value);
			}else{
				WebLogInfo webLogInfoLast = webLogInfoList.get(size-1);
				outPage=webLogInfoLast.getRequestUrl();
				outTime=webLogInfoLast.getRequestTime();
				StringBuffer buffer=new StringBuffer();
				long timeDiff= WebLogParser.timeDiff(webLogInfoLast.getRequestTime(),webLogInfo0.getRequestTime())/1000;
				buffer.append(pageVisits+"\001"+timeDiff+"\001"+outPage+"\001"+outTime+"\001");
				buffer.append(webLogInfo0.toString());
				value.set(buffer.toString());
				context.write(key, value);
			}
			
		}
		
	}
	
	public static void main(String[] args) throws Exception {
		Configuration configuration = new Configuration();
		Job job = Job.getInstance(configuration);
		job.setJarByClass(WebLogVisitSummaryJob.class);

		job.setMapperClass(VisitSummaryMapper.class);
		job.setReducerClass(VisitSummaryReduce.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(WebLogInfo.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		Path in = new Path(args[0]);
		Path out = new Path(args[1]);
		FileSystem fileSystem = FileSystem.get(new URI(in.toString()), new Configuration());
		 if (fileSystem.exists(out)) {
		      fileSystem.delete(out, true);
		  }
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		boolean waitForCompletion = job.waitForCompletion(true);
		System.exit(waitForCompletion ? 0 : 1);
	}
}
